package avicit.bdp.dds.server.worker.task.flink;

import avicit.bdp.common.utils.uploads.FileAdapterUtils;
import avicit.bdp.common.utils.uploads.IFileAdapter;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.bdp.dds.dispatch.process.ResourceInfo;
import avicit.bdp.dds.dispatch.task.AbstractParameters;
import avicit.bdp.dds.dispatch.task.flink.FlinkParameters;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.common.utils.ParameterUtils;
import avicit.bdp.dds.dao.entity.Resource;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.utils.FlinkArgsUtils;
import avicit.bdp.dds.server.utils.ParamUtils;
import avicit.bdp.dds.server.worker.task.AbstractYarnTask;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * flink task
 */
public class FlinkTask extends AbstractYarnTask {

    /**
     * flink command
     */
    private static final String FLINK_COMMAND = "flink";
    private static final String FLINK_RUN = "run";

    /**
     * flink parameters
     */
    private FlinkParameters flinkParameters;

    /**
     * taskExecutionContext
     */
    private TaskExecutionContext taskExecutionContext;


    public FlinkTask(TaskExecutionContext taskExecutionContext, Logger logger) {
        super(taskExecutionContext, logger);
        this.taskExecutionContext = taskExecutionContext;
    }

    @Override
    public void init() {

        logger.info("flink task params {}", taskExecutionContext.getTaskParams());

        flinkParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), FlinkParameters.class);

        if (!flinkParameters.checkParameters()) {
            throw new RuntimeException("flink task params is not valid");
        }
        flinkParameters.setQueue(taskExecutionContext.getQueue());
        setMainJarName();


        if (StringUtils.isNotEmpty(flinkParameters.getMainArgs())) {
            String args = flinkParameters.getMainArgs();


            // replace placeholder
            Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
                    taskExecutionContext.getDefinedParams(),
                    flinkParameters.getLocalParametersMap(),
                    taskExecutionContext.getCmdTypeIfComplement(),
                    taskExecutionContext.getScheduleTime());

            logger.info("param Map : {}", paramsMap);
            if (paramsMap != null) {

                args = ParameterUtils.convertParameterPlaceholders(args, ParamUtils.convert(paramsMap));
                logger.info("param args : {}", args);
            }
            flinkParameters.setMainArgs(args);
        }
    }

    /**
     * create command
     *
     * @return command
     */
    @Override
    protected String buildCommand() {
        List<String> args = new ArrayList<>();

        args.add(FLINK_COMMAND);
        args.add(FLINK_RUN);
        logger.info("flink task args : {}", args);
        // other parameters
        args.addAll(FlinkArgsUtils.buildArgs(flinkParameters));

        String command = ParameterUtils
                .convertParameterPlaceholders(String.join(" ", args), taskExecutionContext.getDefinedParams());

        logger.info("flink task command : {}", command);

        return command;
    }

    @Override
    protected void setMainJarName() {
        // main jar
        ResourceInfo mainJar = flinkParameters.getMainJar();
        if (mainJar != null) {
            String resourceId = mainJar.getId();
            if (resourceId.startsWith(Constants.LEFT_SQUARE_BRACKET)) {
                resourceId = resourceId.substring(2, resourceId.length() - 2);
            }
            String resourceName;
            if (StringUtils.isBlank(resourceId)) {
                resourceName = mainJar.getRes();
            } else {
                Resource resource = processService.getResourceById(flinkParameters.getMainJar().getId());
                if (resource == null) {
                    logger.error("resource id: {} not exist", resourceId);
                    throw new RuntimeException(String.format("resource id: %s not exist", resourceId));
                }
                resourceName = resource.getFullName().replaceFirst(Constants.SINGLE_SLASH, "");
                try {

                    IFileAdapter fileAdapter = FileAdapterUtils.getFileAdapterByProject(taskExecutionContext.getProjectId());
                    fileAdapter.copyServerToLocal(resource.getFullName(),
                            taskExecutionContext.getExecutePath() + resource.getFullName(),
                            false, true);

                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            mainJar.setRes(resourceName);
            flinkParameters.setMainJar(mainJar);
        }
    }

    @Override
    public AbstractParameters getParameters() {
        return flinkParameters;
    }
}
